package csl.blog.server.common;

import com.rabbitmq.client.Channel;
import csl.blog.server.common.model.BaseMessage;
import csl.blog.server.utils.SpringUtils;
import csl.blog.server.utils.JsonMessageConverter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

@Slf4j
public class AsyncEventAccessBuilder<T extends BaseMessage> {

    private final ConnectionFactory connectionFactory = SpringUtils.getBean("connectionFactory");

    public MessageHandle<T> buildTopicMessageSender(final String exchange, final String deadExChange, final String routingKey) throws IOException {
        return buildMessageSender(exchange, deadExChange, routingKey, null, "topic");
    }

    public MessageHandle<T> buildMessageSender(final String exchange, final String deadExChange, final String routingKey, final String queue, final String type) throws IOException {
        Connection connection;
        if (type.equals("direct")) {
            connection = connectionFactory.createConnection();
            buildQueue(exchange, deadExChange, routingKey, queue, connection, "direct");
        } else if (type.equals("topic")) {
            connection = connectionFactory.createConnection();
            buildTopic(exchange, deadExChange, connection);
        }
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setExchange(exchange);
        rabbitTemplate.setRoutingKey(routingKey);
        rabbitTemplate.setMessageConverter(new JsonMessageConverter());
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.info("send message failed: " + cause + correlationData.toString());
            }
        });
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("send message failed: " + replyCode + " " + replyText);
            rabbitTemplate.send(message);
        });
        return message -> {
            try {
                rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(String.valueOf(message.getMsgId())));
            } catch (Exception e) {
                return new AsyncEventRes(false, e.getMessage());
            }
            return new AsyncEventRes(true, "");
        };
    }

    private void buildQueue(String exchange, String deadExChange, String routingKey, final String queue, Connection connection, String type) throws IOException {
        Channel channel = connection.createChannel(false);

        if (type.equals("direct")) {
            channel.exchangeDeclare(exchange, "direct", true, false, null);
        } else if (type.equals("topic")) {
            channel.exchangeDeclare(exchange, "topic", true, false, null);
        }
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 5000);
        args.put("x-dead-letter-exchange", deadExChange);
        args.put("x-dead-letter-routing-key", routingKey + "_dead");
        channel.queueDeclare(queue, true, false, false, args);
        channel.queueBind(queue, exchange, routingKey);
        //声明死信队列
        channel.exchangeDeclare(deadExChange, type, true, false, null);
        channel.queueDeclare(queue + "_dead", true, false, false, null);
        channel.queueBind(queue + "_dead", deadExChange, routingKey + "_dead");
        log.info("exchange:{}, deadExChange:{}, routingKey:{}, queue:{}, type:{}", exchange, deadExChange, routingKey, queue, type);
        try {
            channel.close();
        } catch (TimeoutException e) {
            log.info("close channel time out ", e);
        }
    }

    private void buildTopic(String exchange, String deadExChange, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", deadExChange);
        channel.exchangeDeclare(exchange, "topic", true, false, args);
    }

}
